package org.example.changelog;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class PaimonSamePrimaryKeyTest {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT,"8081");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(conf);
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        TableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String sql1 = "CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'warehouse' = 'file:///tmp/paimon'\n" +
                ");";
        System.out.println(sql1); // 打印SQL语句
        tableEnv.executeSql(sql1);

        // DROP表
        String dropSql = "DROP TABLE if exists `paimon`.`default`.bucket2";
        System.out.println(dropSql); // 打印SQL语句
        tableEnv.executeSql(dropSql);

        // 创建表
        String sql2 = "CREATE TABLE if not exists `paimon`.`default`.bucket2(\n" +
                    "id bigint,\n" +
                    "name String,\n" +
                    "age Int,\n" +
                    "dt string\n" +
//                    "PRIMARY KEY (id) NOT ENFORCED\n" +
                ")\n" +
                "PARTITIONED BY (dt) with\n" +
                "(\n" +
                    "'bucket' = '-1',\n" +
                    "'file.format'='avro',\n" +
                    "'merge-engine'='deduplicate',\n" +
                    "'sink.parallelism' = '2'\n" +
                ");";
        System.out.println(sql2); // 打印SQL语句
        tableEnv.executeSql(sql2);

        String insertSql = "insert into `paimon`.`default`.bucket2 values\n" +
                "(1,'zhangsan',18,'2023-01-01'),\n" +
                "(1,'zhangsan',18,'2023-01-02'),\n" +
                "(1,'zhangsan',18,'2023-01-03');\n";
        System.out.println(insertSql); // 打印SQL语句
        tableEnv.executeSql(insertSql);
    }
}
